package com.atguigu.tingshu.user.service.impl;

import com.alibaba.fastjson.JSON;
import com.atguigu.tingshu.common.constant.KafkaConstant;
import com.atguigu.tingshu.common.constant.SystemConstant;
import com.atguigu.tingshu.common.service.KafkaService;
import com.atguigu.tingshu.common.util.MongoUtil;
import com.atguigu.tingshu.model.user.UserListenProcess;
import com.atguigu.tingshu.user.service.UserListenProcessService;
import com.atguigu.tingshu.vo.album.TrackStatMqVo;
import com.atguigu.tingshu.vo.user.UserListenProcessVo;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Sort;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.Update;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

import java.math.BigDecimal;
import java.util.*;
import java.util.concurrent.TimeUnit;

@Service
@SuppressWarnings({"unchecked", "rawtypes"})
public class UserListenProcessServiceImpl implements UserListenProcessService {

    @Autowired
    private MongoTemplate mongoTemplate;

	@Autowired
	private RedisTemplate redisTemplate;

	@Autowired
	private KafkaService kafkaService;

    @Override
    public Map<String, Object> getLatelyTrack(Long userId) {
        //  声明map 集合
        Map<String,Object> map = new HashMap<>();
        //  获取文档名称
        String collectionName = MongoUtil.getCollectionName(MongoUtil.MongoCollectionEnum.USER_LISTEN_PROCESS,userId);
        //  查询MongoDB 然后按照更新时间;
        Query query = new Query();
        query.with(Sort.by(Sort.Direction.DESC,"updateTime")).limit(1);
        UserListenProcess userListenProcess = this.mongoTemplate.findOne(query, UserListenProcess.class, collectionName);
        //  获取数据.
        map.put("trackId",userListenProcess.getTrackId());
        map.put("albumId",userListenProcess.getAlbumId());
        //  返回数据
        return map;
    }

    @Override
    public void updateListenProcess(Long userId, UserListenProcessVo userListenProcessVo) {
        //	有记录，则更新；没有则新增.
        //	BigDecimal breakSecond = this.getTrackBreakSecond(userListenProcessVo.getTrackId(), userId);
        //	获取集合的名称
        String collectionName = MongoUtil.getCollectionName(MongoUtil.MongoCollectionEnum.USER_LISTEN_PROCESS, userId);
        //	通过集合--找声音Id播放进度.
        //	声明查询条件
        Query query = Query.query(Criteria.where("trackId").is(userListenProcessVo.getTrackId()));
        UserListenProcess userListenProcess = mongoTemplate.findOne(query, UserListenProcess.class, collectionName);
        //	判断这个对象
        if (null == userListenProcess) {
            //	创建对象
            userListenProcess = new UserListenProcess();
            //	赋值：
            BeanUtils.copyProperties(userListenProcessVo, userListenProcess);
            userListenProcess.setUserId(userId);
            userListenProcess.setIsShow(0);
            userListenProcess.setCreateTime(new Date());
            userListenProcess.setUpdateTime(new Date());
            //	保存到MongoDB中.
            this.mongoTemplate.save(userListenProcess, collectionName);
        } else {
            //	第n次播放了。
            Update update = new Update();
            update.set("breakSecond", userListenProcessVo.getBreakSecond());
            update.set("updateTime",new Date());
            this.mongoTemplate.upsert(query, update, UserListenProcess.class, collectionName);
            //	根据trackId 修改 breakSecond
            //		userListenProcess.setBreakSecond(userListenProcessVo.getBreakSecond());
            //		//	修改：
            //		this.mongoTemplate.save(userListenProcess,collectionName);
        }
        //	记录声音播放次数。同一天，同一个用户，无论播放多少次。只+1次.
        //	redis setnx key value; 当key 不存在的时候生效.
        //	自定义一个key
        String key = userId + ":" + userListenProcessVo.getTrackId();
		//	setnx+setex = set lock ok ex 5 nx具有原子性;
		Boolean result = this.redisTemplate.opsForValue().setIfAbsent(key, userId, 24, TimeUnit.HOURS);
		//	判断
		if (result){
			//	执行成功. 表示要更新声音播放次数.
			//	将更新数据的条件封装到实体类中
			TrackStatMqVo trackStatMqVo = new TrackStatMqVo();
			trackStatMqVo.setTrackId(userListenProcessVo.getTrackId());
			trackStatMqVo.setStatType(SystemConstant.TRACK_STAT_PLAY);
			trackStatMqVo.setCount(1);
			trackStatMqVo.setAlbumId(userListenProcess.getAlbumId());
			//	业务编号：防止消息重复消费的.
			String businessNo = UUID.randomUUID().toString();
			trackStatMqVo.setBusinessNo(businessNo);
			//	发送消息异步执行. update track_stat set stat_num = ? where stat_typ=? track_id=?
			kafkaService.sendMessage(KafkaConstant.QUEUE_TRACK_STAT_UPDATE, JSON.toJSONString(trackStatMqVo));
		}
	}

    @Override
    public BigDecimal getTrackBreakSecond(Long trackId, Long userId) {
        //	获取集合的名称
        String collectionName = MongoUtil.getCollectionName(MongoUtil.MongoCollectionEnum.USER_LISTEN_PROCESS, userId);
        //	通过集合--找声音Id播放进度.
        //	声明查询条件
        Query query = Query.query(Criteria.where("trackId").is(trackId));
        UserListenProcess userListenProcess = mongoTemplate.findOne(query, UserListenProcess.class, collectionName);
        //	判断对象是否为空
        if (null != userListenProcess) {
            //	返回数据
            return userListenProcess.getBreakSecond();
        }
        //	默认从头开始播放
        return new BigDecimal("0");
    }
}
